package com.doit.day02;

import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //必须得设置的
        //kafka集群的连接地址
        props.setProperty("bootstrap.servers","linux01:9092,linux02:9092,linux03:9092");
        //key的序列化方式
//        org.apache.kafka.common.serialization.Serializer
//        props.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //value的序列化方式
        props.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        //可以选配的
        //kafka的应答机制
        props.setProperty("acks","-1");

        //获取对象  获取生产者的对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 0; i < 10; i++) {
            //用对象操作api   造数据
            ProducerRecord<String, String> record = new ProducerRecord<>("java", "heheehehhe"+i);
            //用对象send数据到kafka中
            producer.send(record);
            //刷新一下
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(100,300));
        }
        //关闭资源
        producer.close();

    }
}
